package kim.artemis.rpc.core.protocol.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import kim.artemis.rpc.core.model.ArtemisRpcRequest;
import kim.artemis.rpc.core.model.ArtemisRpcResponse;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 服务端处理程序
 * @author Wq.
 */
@Slf4j
public class ArtemisRpcServerHandler extends SimpleChannelInboundHandler<ArtemisRpcRequest> {

    /**
     * 服务端处理线程池
     */
    private ThreadPoolExecutor poolExecutor;

    /**
     * 调用处理器
     */
    private ArtemisRpcServerInvoker invokerHandler;

    public ArtemisRpcServerHandler(ArtemisRpcServerInvoker invokerHandler, ThreadPoolExecutor poolExecutor) {
        this.invokerHandler = invokerHandler;
        this.poolExecutor = poolExecutor;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ArtemisRpcRequest artemisRpcRequest) {
        try {
            poolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    ArtemisRpcResponse artemisRpcResponse = invokerHandler.invoker(artemisRpcRequest);
                    channelHandlerContext.writeAndFlush(artemisRpcResponse);
                }
            });
        } catch (Exception e) {
            ArtemisRpcResponse artemisRpcResponse = new ArtemisRpcResponse();
            artemisRpcResponse.setArtemisId(artemisRpcRequest.getArtemisId());
            artemisRpcResponse.setException(e.getLocalizedMessage());
            channelHandlerContext.writeAndFlush(artemisRpcResponse);
        }
    }

    @SneakyThrows
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error(">>>>>>> artemis rpc, server finds an exception: {}", cause.getLocalizedMessage());
        log.error(">>>>>>> artemis rpc, the address of the closed connection is: {}", ctx.channel().remoteAddress());
        ctx.close();
    }
}
